package com.hadoop.dao;

import java.io.Closeable;
import java.io.File;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.OutputStream;
import java.net.URI;
import java.util.List;

import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;
import org.apache.log4j.Logger;

import com.google.common.collect.Lists;
import com.hadoop.dao.exception.HdfsDaoException;

import static com.hadoop.dao.util.CheckUtil.*;

/**
 * HdfsDao
 * 
 * @author yangshenhui
 */
public class HdfsDao {
	private static final String HDFS = "hdfs://redhat:9000/";
	private static Logger logger = Logger.getLogger(HdfsDao.class);
	private String hdfsPath;
	private Configuration configuration;

	public HdfsDao(Configuration configuration) {
		this(HDFS, configuration);
	}

	public HdfsDao(String hdfsPath, Configuration configuration) {
		this.hdfsPath = hdfsPath;
		this.configuration = configuration;
	}

	public boolean mkdir(String directory) throws HdfsDaoException {
		checkEmptyString(directory);
		logger.info(String.format("start creating hdfs directory %s.",
				directory));
		FileSystem fileSystem = null;
		try {
			fileSystem = FileSystem.get(URI.create(hdfsPath), configuration);
			Path path = new Path(directory);
			if (fileSystem.exists(path)) {
				logger.info(String
						.format("hdfs directory %s already exists, you can not create.",
								directory));
				return false;
			}
			fileSystem.mkdirs(path);
			logger.info(String.format(
					"hdfs directory %s created successfully.", directory));
			logger.info(String.format("end creating hdfs directory %s.",
					directory));
		} catch (IOException e) {
			throw new HdfsDaoException(e);
		} finally {
			close(fileSystem);
		}
		return true;
	}

	public boolean rmr(String directory) throws HdfsDaoException {
		checkEmptyString(directory);
		logger.info(String.format("start deleting hdfs directory %s.",
				directory));
		FileSystem fileSystem = null;
		try {
			fileSystem = FileSystem.get(URI.create(hdfsPath), configuration);
			Path path = new Path(directory);
			fileSystem.deleteOnExit(path);
			logger.info(String.format(
					"hdfs directory %s deleted successfully.", directory));
			logger.info(String.format("end deleting hdfs directory %s.",
					directory));
		} catch (IOException e) {
			throw new HdfsDaoException(e);
		} finally {
			close(fileSystem);
		}
		return true;
	}

	public List<Path> ls(String directory) throws HdfsDaoException {
		checkEmptyString(directory);
		logger.info(String.format("start hdfs directory %s lists.", directory));
		FileSystem fileSystem = null;
		List<Path> pathList = null;
		try {
			fileSystem = FileSystem.get(URI.create(hdfsPath), configuration);
			Path path = new Path(directory);
			if (!fileSystem.exists(path)) {
				logger.info(String.format("hdfs directory %s does not exist.",
						directory));
				return null;
			}
			pathList = Lists.newArrayList();
			ls(pathList, fileSystem, path);
			logger.info(String
					.format("end hdfs directory %s lists.", directory));
		} catch (IOException e) {
			throw new HdfsDaoException(e);
		} finally {
			close(fileSystem);
		}
		return pathList;
	}

	private void ls(List<Path> fileList, FileSystem fileSystem, Path path)
			throws FileNotFoundException, IOException {
		FileStatus[] fileStatuses = fileSystem.listStatus(path);
		for (FileStatus fileStatus : fileStatuses) {
			if (fileStatus.isDirectory()) {
				ls(fileList, fileSystem, fileStatus.getPath());
			} else {
				fileList.add(fileStatus.getPath());
			}
		}
	}

	public boolean create(String fileName, List<String> contentList)
			throws HdfsDaoException {
		checkEmptyString(fileName);
		logger.info(String.format("start creating hdfs file %s.", fileName));
		FileSystem fileSystem = null;
		FSDataOutputStream fsDataOutputStream = null;
		try {
			fileSystem = FileSystem.get(URI.create(hdfsPath), configuration);
			Path path = new Path(fileName);
			if (fileSystem.exists(path)) {
				logger.info(String.format(
						"hdfs file %s already exists, you can not create.",
						fileName));
				return false;
			}
			fsDataOutputStream = fileSystem.create(path);
			if (isNotEmpty(contentList)) {
				for (String content : contentList) {
					byte[] buffer = content.getBytes();
					fsDataOutputStream.write(buffer, 0, buffer.length);
					fsDataOutputStream.write("\n".getBytes());
				}
			}
			logger.info(String.format("end creating hdfs file %s.", fileName));
		} catch (IOException e) {
			throw new HdfsDaoException(e);
		} finally {
			close(fsDataOutputStream);
			close(fileSystem);
		}
		return true;
	}

	public boolean upload(String localFile, String remoteDirectory)
			throws HdfsDaoException {
		checkEmptyString(localFile);
		checkEmptyString(remoteDirectory);
		logger.info("hdfs start uploading files to the directory.");
		FileSystem fileSystem = null;
		try {
			fileSystem = FileSystem.get(URI.create(hdfsPath), configuration);
			fileSystem.copyFromLocalFile(new Path(localFile), new Path(
					remoteDirectory));
			logger.info(String.format(
					"upload local file %s to hdfs directory %s.", localFile,
					remoteDirectory));
			logger.info("the end of the directory to upload files to hdfs.");
		} catch (IOException e) {
			throw new HdfsDaoException(e);
		} finally {
			close(fileSystem);
		}
		return true;
	}

	public boolean download(String remoteFile, String localDirectory)
			throws HdfsDaoException {
		checkEmptyString(remoteFile);
		checkEmptyString(localDirectory);
		logger.info("hdfs file to start the download.");
		FileSystem fileSystem = null;
		FSDataInputStream fsDataInputStream = null;
		OutputStream outputStream = null;
		try {
			fileSystem = FileSystem.get(URI.create(hdfsPath), configuration);
			fsDataInputStream = fileSystem.open(new Path(remoteFile));
			localDirectory = localDirectory + File.separator
					+ extractFileName(remoteFile);
			outputStream = new FileOutputStream(new File(localDirectory));
			IOUtils.copyBytes(fsDataInputStream, outputStream, 4096, false);
			logger.info(String.format(
					"download hdfs file %s to a local directory %s.",
					remoteFile, localDirectory));
			logger.info("end hdfs file download.");
		} catch (IOException e) {
			throw new HdfsDaoException(e);
		} finally {
			close(outputStream);
			close(fsDataInputStream);
			close(fileSystem);
		}
		return true;
	}

	private String extractFileName(String filePath) {
		return StringUtils.substring(filePath,
				StringUtils.lastIndexOf(filePath, "/") + 1, filePath.length());
	}

	private void close(Closeable closeable) throws HdfsDaoException {
		if (isNotNull(closeable)) {
			try {
				closeable.close();
			} catch (IOException e) {
				throw new HdfsDaoException(e);
			}
		}
	}

}